package cn.doitedu.datayi.utils

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.{DataFrame, SparkSession}

import java.util.Properties

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-09-22
 * @desc gps参考数据转geohash参考数据
 */
object GpsGeoHash {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .appName("geohash编码")
      .config("spark.sql.shuffle.partitions","2")
      .master("local")
      .enableHiveSupport()  // 开启hive支持
      .getOrCreate()

    // 读取mysql中的gps地点参考表（已经被扁平化的表）
    val props = new Properties()
    props.load(GpsGeoHash.getClass.getClassLoader.getResourceAsStream("mysql.properties"))

    val gpsTable: DataFrame = spark.read.jdbc(props.getProperty("url"), "ref_area_gps", props)
    gpsTable.createTempView("gps")



    // 把经纬度转成geohash码
    val gps2geo = (lat:Double,lng:Double)=>
      GeoHash.geoHashStringWithCharacterPrecision(lat,lng,6)

    // 注册函数
    spark.udf.register("gps2geo",gps2geo)

    // 写sql来将源数据中的gps坐标转成geohash编码
    spark.sql(
      """
        |create table dim.geohash_area
        |as
        |select
        |  gps2geo(lat,lng) as geohash,
        |  province,
        |  city,
        |  region
        |from gps
        |
        |""".stripMargin)

    spark.close()
  }
}
